package com.example.kafka.service;

import com.example.kafka.config.KafkaClientManager;
import com.example.kafka.model.Order;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.errors.TimeoutException;
import org.apache.kafka.common.errors.NetworkException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import java.util.UUID;
import java.time.LocalDateTime;
import java.math.BigDecimal;
import java.util.concurrent.ExecutionException;

/**
 * 订单服务类
 */
@Service
public class OrderService {

    private static final Logger logger = LoggerFactory.getLogger(OrderService.class);
    private static final String ORDER_TOPIC = "order-topic";

    @Autowired
    private KafkaClientManager kafkaClientManager;

    /**
     * 创建订单
     */
    public Order createOrder(String userId, double amount) {
        Order order = new Order();
        order.setOrderId(UUID.randomUUID().toString());
        order.setUserId(userId);
        order.setAmount(new BigDecimal(amount));
        order.setStatus("CREATED");
        order.setCreateTime(LocalDateTime.now());

        try {
            // 获取当前 Kafka 客户端
            KafkaProducer<String, String> producer = kafkaClientManager.getProducer();

            // 发送订单消息到 Kafka
            ProducerRecord<String, String> record = new ProducerRecord<>(ORDER_TOPIC, order.getOrderId(), order.toString());
            producer.send(record).get(); // 同步发送，等待结果

            logger.info("消息发送成功，当前 Kafka 客户端数量: {}", 0);
        } catch (TimeoutException | NetworkException e) {
            // 直接捕获 Kafka 服务端连接异常
            logger.error("Kafka 服务端连接异常，尝试重新创建 Kafka 客户端", e);
            kafkaClientManager.recreateProducer();
        } catch (ExecutionException e) {
            // 其他异常只记录日志
            logger.error("发送消息失败", e);
        } catch (InterruptedException e) {
            // 线程中断异常只记录日志
            logger.error("发送消息被中断", e);
            Thread.currentThread().interrupt();
        }

        return order;
    }

    /**
     * 查询订单
     */
    public Order getOrder(String orderId) {
        // 这里模拟从数据库查询订单
        Order order = new Order();
        order.setOrderId(orderId);
        order.setUserId("user123");
        order.setAmount(new BigDecimal("100.00"));
        order.setStatus("PAID");
        order.setCreateTime(LocalDateTime.now());
        return order;
    }
}
